package com.cn.task;

import java.io.BufferedInputStream;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.FileReader;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Future;

import net.sf.json.JSONArray;
import net.sf.json.JSONObject;

import org.apache.commons.io.IOUtils;
import org.springframework.transaction.TransactionDefinition;

import com.cn.bean.SimpleList;
import com.cn.bean.SimpleMap;
import com.cn.dao.RowSet;
import com.cn.dao.SqlParam;
import com.cn.dao.TransactionManager;
import com.cn.exception.MyException;
import com.cn.ftp.FtpFactory;
import com.cn.ftp.FtpUtil;
import com.cn.task.beans.FtpConfigBean;
import com.cn.task.beans.TaskConfigBean;
import com.cn.task.timer.IScheduleIterator;
import com.cn.xc.utils.BeanUtil;
import com.cn.xc.utils.CollectionUtil;
import com.cn.xc.utils.StringUtil;


/**
 * 
 * @author Leezy Description ：
 * 
 *         modified by sushy
 */
public class DefaultTimerTask extends BaseTask {

	String taskCode;
	TaskConfigBean taskConfig;
	FtpConfigBean ftpConfig;
	AtomicIntSeq fileSeq = new AtomicIntSeq(99999999);

	public String getTaskCode() {
		return taskCode;
	}

	public void setTaskCode(String taskCode) {
		this.taskCode = taskCode;
	}

	public TaskConfigBean getTaskConfig() {
		return taskConfig;
	}

	public void setTaskConfig(TaskConfigBean taskConfig) {
		this.taskConfig = taskConfig;
	}

	public FtpConfigBean getFtpConfig() {
		return ftpConfig;
	}

	public void setFtpConfig(FtpConfigBean ftpConfig) {
		this.ftpConfig = ftpConfig;
	}

	@SuppressWarnings("unchecked")
	@Override
	public void loadTaskConf() {
		fileSeq.set(1);
		logger.warn("---TimerTask init begin..." + taskCode + "\n");
		//#System.out.println("---TimerTask init begin..." + taskCode + "\n");
		if (StringUtil.isEmpty(taskCode)) {
			logger.error("---The Task " + taskCode + ":运行异常! Spring注入taskCode配置为空!");
			throw new MyException("任务缺少配置，Spring注入taskCode配置为空!");
		}

		TaskConfigBean taskConfig = new TaskConfigBean();
		try {
			SqlParam sqlParam = new SqlParam();
			sqlParam.addParam("taskCode", taskCode);
			sqlParam.addSql("select PARAM_CODE, PARAM_VALUE from TB_IC_TASK_CONF where STATE = 'S0A' and TASK_CODE = :taskCode");
			RowSet rs = jdbcDao.query(sqlParam);
			while (rs.next()) {
				this.saveToBean(rs, taskConfig);
			}
		} catch (Exception e) {
			logger.error("---The Task " + taskCode + ":read config information error!", e);
			throw new MyException("读取任务配置数据异常!" + e.getMessage());
		}

		String ruleCode = StringUtil.trim(taskConfig.getQrySql());
		if (StringUtil.isNotEmpty(ruleCode)) {
			try {
				SqlParam sqlParam = new SqlParam();
				sqlParam.addParam("ruleCode", ruleCode);
				sqlParam.addSql("select RULE_SQL from ctg_qry_rule where RULE_TYPE = 'SELECT' and STATE = 'S0A' and rule_code = :ruleCode order by rule_index ");

				RowSet rs = jdbcDao.query(sqlParam);
				while (rs.next()) {
					String sql = rs.getString(1);
					if (StringUtil.isNotEmpty(StringUtil.trim(sql))) {
						taskConfig.addQrySqls(sql);
					}
				}
			} catch (Exception e) {
				logger.error("---The Task " + taskCode + ":read the sql config error!", e);
				throw new MyException("读取任务配置数据查询语句异常!" + e.getMessage());
			}
		}

		ruleCode = StringUtil.trim(taskConfig.getAfterSql());
		if (StringUtil.isNotEmpty(ruleCode)) {
			try {
				SqlParam sqlParam = new SqlParam();
				sqlParam.addParam("ruleCode", ruleCode);
				sqlParam.addSql("select RULE_SQL from ctg_qry_rule where RULE_TYPE = 'SELECT' and STATE = 'S0A' and rule_code = :ruleCode order by rule_index ");

				RowSet rs = jdbcDao.query(sqlParam);
				while (rs.next()) {
					String sql = rs.getString(1);
					if (StringUtil.isNotEmpty(StringUtil.trim(sql))) {
						taskConfig.addAfterSqls(sql);
					}
				}
			} catch (Exception e) {
				logger.error("---The Task " + taskCode + ":error after read the config ", e);
				throw new MyException("读取任务配置数据查询语句异常!" + e.getMessage());
			}
		}

		this.setTaskConfig(taskConfig);

		setTaskName(taskConfig.getTaskName());
		setIteratorParam(taskConfig.getIteratorParam());
		setPoolName(taskConfig.getPoolName());

		try {
			setIteratorClass((Class<IScheduleIterator>) Class.forName(taskConfig.getIteratorClass()));
		} catch (ClassNotFoundException e) {
			logger.error("---The Task " + taskCode + ":找不到对应配置定时类" + taskConfig.getIteratorClass());
			throw new MyException( "读取任务配置数据查询语句异常!" + e.getMessage());
		} catch (Throwable t) {
			logger.error("---The Task " + taskCode + ":设置任务定时类失败" + taskConfig.getIteratorClass());
			throw new MyException("设置定时任务类型异常!" + t.getMessage());
		} finally {

		}
		
		//#文件SEQ_ID的进程重启是否延续上次已经生成文件的处理
		String seqType = taskConfig.getSeqType();
		if ("continueByDay".equals(seqType)) {
			//#每天重置，重启根据已经生成文件续传

			int seqId = 0;
			String fileCheckCons = taskConfig.getFileCheckCons();
			SimpleMap fileParam = new SimpleMap();
			fileParam.putAll((HashMap<String, Object>) BeanUtil.toMap(taskConfig));
			String fileName = CtgFileService.getFileName(fileCheckCons, fileParam);
			String filePath = CtgFileService.genFilePath(taskConfig.getSubPath());
			File folder = new File(filePath);
			if (!folder.exists()) {
				return;
			}
			File[] list = folder.listFiles();
			for (File file : list) {
				if (!file.isFile()) {
					continue;
				}

				String curFileName = file.getName();
				if (curFileName.length() > fileName.length()) {
					if (curFileName.startsWith(fileName)) {
						String suffix = curFileName.substring(fileName.length());
						if (suffix.indexOf(".") > 0) {
							String curSeq = suffix.substring(0, suffix.indexOf("."));
							if (Integer.valueOf(curSeq) > seqId) {
								seqId = Integer.valueOf(curSeq);
							}
						}
					}
				}
			}

			if (seqId > 0) {
				fileSeq.set(seqId + 1);
			}

		}//## continueByDay - end
		logger.warn("---TimerTask init end..." + taskCode + "\n");
	}

	@Override
	public void run() {
		if (isRunning) {
			return;
		}
		isRunning = true;

		try {
			logger.warn("---The Task " + taskCode + ":================ Task Start ==============");
			doRun();
			logger.warn("---The Task " + taskCode + ":================ Task End ==============");
		} catch (Throwable t) {
			isRunning = false;
			logger.error("---The Task " + taskCode + ":运行异常!" + t.getMessage(), t);
		}
		isRunning = false;
	}

	public void doRun() {
		String filePath = "";
		long tatalRownum = 0;
		List<String> fileList = new ArrayList<String>();

		filePath = CtgFileService.genFilePath(taskConfig.getSubPath());
		taskConfig.setFilePath(filePath);
		String rowSep = "\r\n";
		taskConfig.setRowSeparator(rowSep);

		String fileNames = taskConfig.getFileNameCons();
		//## 内容为空时，是否必传
		String isMust = isMustUpload(fileNames);
		String isMustUpload = null;
		if (isMust != null && (!"$notMust$".equalsIgnoreCase(isMust))) {//## 必传
			isMustUpload = isMust;
		}
		SimpleMap variableNames = getVariableName(fileNames);
		//## go to see function getVariableName() note
		if (variableNames == null || variableNames.size() < 1) {
			logger.error("---The Task " + taskCode + ":运行时错误 ! 缺少fileNameCons下的variableName的配置，至少有一个值!");
			throw new MyException("缺少fileNameCons下的variableName的配置，至少有一个值!");
			//return;
		}
		System.out.println("variableName: " + variableNames);
		for (Map.Entry<String, Object> entry : variableNames.entrySet()) {
			String key = entry.getKey();
			String replaceName = String.valueOf(entry.getValue());
			System.out.println(key + "=" + replaceName);

			SimpleMap fileParam = new SimpleMap();
			fileParam.put("isMustUpload", isMustUpload);
			fileParam.put("replaceName", replaceName);
			String platCode = replaceName;
			fileParam.put("platCode", platCode);
			fileParam.putAll((SimpleMap) BeanUtil.toMap(taskConfig));
			fileParam.put("needFileSeq", "no");
			List<Future<Object>> listFuture = new ArrayList<Future<Object>>();
			//#生成本地网使用到的配置信息
			SimpleMap curFileParam = new SimpleMap();
			curFileParam.put("isMustUpload", isMustUpload);
			curFileParam.put("replaceName", replaceName);
			curFileParam.put("platCode", platCode);
			curFileParam.putAll((SimpleMap) BeanUtil.toMap(taskConfig));
			curFileParam.put("taskCode", taskCode);

			Future<Object> future = asyncDoTask(curFileParam, true);
			if (future != null) {
				listFuture.add(future);
			} else {
				//#单线程的数据，已经放fileList里面
			}
			//#等待所有请求都处理完

			logger.warn("---The Task " + taskCode + ":异步文件处理启动完毕!");

			for (Future<Object> f : listFuture) {
				try {
					SimpleMap resultMap = (SimpleMap) getFuture(f);
					long rownum = resultMap.getLong("rownum");
					tatalRownum += rownum;
					fileList.add(resultMap.getString("absoluteFile"));
					logger.warn("---The Task " + taskCode + ":生成中间文件成功：\r\n--------"
							+ resultMap.getString("absoluteFile") + "\r\n--------deal line sum="
							+ String.valueOf(rownum));

				} catch (Throwable t) {
					logger.error("---获取处理结果失败!", t);
				}
			}

			logger.warn("---The Task " + taskCode + ":异步文件处理结果查询完毕!");

			FileOutputStream fos = null;
			OutputStreamWriter osw = null;
			List<String> finalFiles = new ArrayList<String>();
			
			try {
				//#文件合并，生成最终文件
				String fileNameCons = taskConfig.getFileNameCons();
				fileParam.put("fileSeq", fileSeq);

				String splitFormat = taskConfig.getFileSplit();
				long lineLimit = taskConfig.getLineLimit();
				if ("splitByLine".equals(splitFormat) && lineLimit > 0) {
					int curLine = 0; //#当前处理文件已经写入记录数
					long oddNum = tatalRownum; //#剩余未处理的文件记录数总数
					//#把文件合到目标文件后删除临时文件
					for (String fileLocal : fileList) {

						if (StringUtil.isEmpty(fileLocal)) {
							continue;
						}
						File cityFile = new File(fileLocal);
						if (!cityFile.exists()) {
							continue;
						}

						FileReader reader = new FileReader(fileLocal);
						BufferedReader bReader = new BufferedReader(reader);
						String oneLine = "";
						while ((oneLine = bReader.readLine()) != null) {
							if (curLine == 0) {
								//#创建目标文件
								fileParam.put("needFileSeq", "yes");
								String fileName = CtgFileService.getFileName(fileNameCons, fileParam);
								finalFiles.add(fileName);
								File desFile = CtgFileService.getFile(filePath, fileName);
								CtgFileService.createNewFile(desFile);
								//#生成第一第二行
								fos = new FileOutputStream(desFile);
								osw = new OutputStreamWriter(fos, taskConfig.getCharsetName());

								String firstLine = "";
								if (oddNum - lineLimit > 0) {
									firstLine = String.valueOf(lineLimit);
									oddNum = oddNum - lineLimit;
								} else {
									firstLine = String.valueOf(oddNum);
								}

								String secondLine = taskConfig.getColumnNames().replace(",",
										taskConfig.getColSeparator());
								String columoNames = taskConfig.getColumnNames();
								//## 文件内容第1,2行不需要写入记录数和列名
								if (StringUtil.isNotEmpty(columoNames)) {
									osw.write(firstLine);
									osw.write(rowSep);
									osw.write(secondLine);
									osw.write(rowSep);
									osw.flush();
								}
							}

							osw.write(oneLine);
							osw.write(rowSep);
							curLine++;
							//#作为标识，换行处理
							if (curLine >= lineLimit) {
								osw.flush();
								IOUtils.closeQuietly(osw);
								curLine = 0;
							}
						}

						bReader.close();
						reader.close();
						cityFile.delete();

					}
				} else {
					//#创建目标文件
					String fileName = CtgFileService.getFileName(fileNameCons, fileParam);
					File file = CtgFileService.getFile(filePath, fileName);

					CtgFileService.createNewFile(file);
					//#生成第一第二行
					fos = new FileOutputStream(file);
					osw = new OutputStreamWriter(fos, taskConfig.getCharsetName());
					String firstLine = String.valueOf(tatalRownum);
					String secondLine = taskConfig.getColumnNames().replace(",", taskConfig.getColSeparator());
					String columoNames = taskConfig.getColumnNames();
					//## 文件内容第1,2行不需要写入记录数和列名
					if (StringUtil.isNotEmpty(columoNames)) {
						osw.write(firstLine);
						osw.write(rowSep);
						osw.write(secondLine);
						osw.write(rowSep);
						osw.flush();
					}
					//#把文件合到目标文件后删除临时文件
					for (String fileLocal : fileList) {

						if (StringUtil.isEmpty(fileLocal)) {
							continue;
						}

						File cityFile = new File(fileLocal);
						if (!cityFile.exists()) {
							continue;
						}

						BufferedInputStream in = new BufferedInputStream(new FileInputStream(cityFile));
						IOUtils.copy(in, osw);
						osw.flush();
						in.close();

						cityFile.delete();
					}
					finalFiles.add(fileName);
				}

			} catch (IOException ex) {
				ex.printStackTrace();
				return;
			} catch (Exception e) {
				e.printStackTrace();
				return;
			} finally {
				IOUtils.closeQuietly(osw);
				IOUtils.closeQuietly(fos);
			}

			logger.warn("---The Task " + taskCode + ":目标上传文件生成完毕!");
			if (finalFiles.size() == 0) {
				logger.warn("---The Task " + taskCode + ":no data, not need to upload. Happy job");
				return;
			}

			for (String curFile : finalFiles) {
				logger.warn("---The Task " + taskCode + ":目标文件生成成功：" + curFile);
			}

			FtpUtil ftp = FtpFactory.getFtpUtil(taskConfig.getFtpConfig());

			if (ftp == null) {
				logger.warn("---The Task " + taskCode + ":can not read the ftp config of the task.");
				return;
			}

			try {
				//#连接服务器
				ftp.connectServer();
				for (String curFile : finalFiles) {
					//#上传文件
					ftp.upload(filePath + "/" + curFile, curFile);
					logger.warn("---The Task " + taskCode + ":文件上传成功：" + curFile);
				}

			} catch (Exception e) {
				logger.error("---The Task " + taskCode + "文件上传出错！", e);
				e.printStackTrace();
				return;
			} finally {
				try {
					ftp.closeServer(); //#关闭服务器连接
					//## logger.warn("The Task " + taskCode + ":文件上传完毕!");
				} catch (Exception e) {
					logger.error("---The Task " + taskCode + ":关闭FTP连接出错!", e);
					e.printStackTrace();
				}
			}

			try {
				String doAfter = taskConfig.getDoAfter();
				if ("need".equals(doAfter) && CollectionUtil.isNotEmpty(taskConfig.getAfterSqls())) {
					//# 上传成功时，还需要进一步处理，则写在这里
				}
			} finally {
			}
		}
	}

	public void doAterByCityId(String cityId, List<String> execSqls) {
		TransactionManager trans = new TransactionManager(jdbcDao.getDataSource(),
				TransactionDefinition.PROPAGATION_REQUIRES_NEW);
		try {
			trans.startTransaction();

			for (String sql : execSqls) {
				jdbcDao.execSql(sql);
			}

			trans.commit();
		} finally {
			trans.rollback();
		}

	}

	/**
	 * //# 获取可变文件名替换
	 * 
	 * warn： DefaultTimerTask新逻辑处理下，为了兼容，variableName必须有一个值或者多个！
	 * variableName的值按键值对的形式配置 文件名固定时,variableName有一个值即可
	 * 
	 * @param fileNameCons
	 *            配置串
	 * @return
	 */
	private SimpleMap getVariableName(String fileNameCons) {
		String[] fileCons = StringUtil.splitTokens(fileNameCons, ",");
		for (String con : fileCons) {
			if (con.length() > 13 && "variableName".equalsIgnoreCase(con.substring(0, 12))) {
				String[] names = StringUtil.splitTokens(con, "~");
				if (names.length > 1) {
					return parseJSON2Map(names[1]);
				}
			}
		}

		return new SimpleMap();
	}

	/**
	 * //# 当文件内容为空时，是否必须上传，是则写入mustUpload指定的配置
	 * 
	 * @param fileNameCons
	 * @return 默认写入的内容
	 */
	private String isMustUpload(String fileNameCons) {
		String[] fileCons = StringUtil.splitTokens(fileNameCons, ",");
		for (String con : fileCons) {
			if (con.length() > 11 && "mustUpload".equalsIgnoreCase(con.substring(0, 10))) {
				String[] conts = StringUtil.splitTokens(con, "~");
				if (conts.length > 1) {
					System.out.println("内容为空且必须上传时，默认写入：" + conts[1]);
					return conts[1].toString();
				}
			}
		}

		return "$notMust$";
	}

	/**
	 * //# json => map
	 * 
	 * @param jsonStr
	 * @return
	 */
	@SuppressWarnings("unchecked")
	private SimpleMap parseJSON2Map(String jsonStr) {
		if (StringUtil.isNotEmpty(jsonStr)) {
			SimpleMap map = new SimpleMap();
			jsonStr = jsonStr.replace("##", ",");
			JSONObject json = JSONObject.fromObject(jsonStr);
			for (Object key : json.keySet()) {
				Object value = json.get(key);
				//# analse next json
				if (value instanceof JSONArray) {
					SimpleList list = new SimpleList();
					Iterator<JSONObject> it = ((JSONArray) value).iterator();
					while (it.hasNext()) {
						JSONObject js = it.next();
						list.add(parseJSON2Map(js.toString()));
					}
					map.put(key.toString(), list);
				} else {
					map.put(key.toString(), value);
				}
			}
			return map;
		}

		return new SimpleMap();
	}

}